package com.tjs.lucky.application.worker;

import com.db.router.strategy.IDBRouterStrategy;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.lucky.tjs.annotation.ElasticJobConfig;
import com.tjs.lucky.application.mq.RabbitMQProducer;
import com.tjs.lucky.common.MQState;
import com.tjs.lucky.domain.activity.model.vo.DisAwardInvoiceVO;
import com.tjs.lucky.domain.activity.service.partake.IActivityPartake;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.List;

import static com.tjs.lucky.common.Constants.DIS_AWARD_QUEUE_NAME;

/**
 * @author tianjiashu（mac）
 * @ClassName LuckyOrderMQStateJob.java
 * @Description TODO: 如果一直失败怎么办？corn表达式是临时的，需要改
 * @createTime 2023年06月20日 21:29:00
 */
@Component
@Slf4j
@ElasticJobConfig(jobName = "LuckyOrderMQStateJob", cron = "0 0 0 * * ? *", shardingTotalCount = 4)
public class LuckyOrderMQStateJob implements SimpleJob {

    @Resource
    private IActivityPartake activityPartake;

    @Resource
    private IDBRouterStrategy dbRouter;

    @Resource
    private RabbitMQProducer rabbitMQProducer;

    @Override
    public void execute(ShardingContext shardingContext) {

        int tbNumber = shardingContext.getShardingItem();
        int dbCount = dbRouter.tbCount();
        for (int dbNumber = 0; tbNumber < dbCount; tbNumber++) {
            // 扫描库表数据
            List<DisAwardInvoiceVO> disAwardInvoiceVOList = activityPartake.scanInvoiceMqState(dbNumber, tbNumber);
            log.info("扫描用户抽奖奖品发放MQ状态[Table = 2*4] 扫描库：{} 扫描表：{} 扫描数：{}", dbNumber, tbNumber, disAwardInvoiceVOList.size());

            // 补偿 MQ 消息  TODO: 次数限制
            for (DisAwardInvoiceVO disAwardInvoiceVO : disAwardInvoiceVOList) {
                rabbitMQProducer.sendMessageToLuckyExchange(
                        DIS_AWARD_QUEUE_NAME,
                        disAwardInvoiceVO,
                        "award",
                        new ListenableFutureCallback<CorrelationData.Confirm>() {
                            @Override
                            public void onFailure(Throwable e) {
                                activityPartake.updateInvoiceMqState(disAwardInvoiceVO.getUId(), disAwardInvoiceVO.getOrderId(), MQState.PRODUCT_FAIL.getCode());
                                log.info("异步发奖：消息发送失败。原因：{}", e.getMessage());
                            }

                            @Override
                            public void onSuccess(CorrelationData.Confirm confirm) {
                                activityPartake.updateInvoiceMqState(disAwardInvoiceVO.getUId(), disAwardInvoiceVO.getOrderId(), MQState.COMPLETE.getCode());
                                log.info("异步发奖：消息发送成功!");
                            }
                        }
                );
            }
        }
    }
}
